package com.bingxu.flink.action;

import com.bingxu.flink.bean.AdsClickLog;
import com.bingxu.flink.bean.OrderEvent;
import com.bingxu.flink.bean.TxEvent;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class Order_OnTime_Monit {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        // load data and transform Java Bean object
        DataStreamSource<String> orderDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\OrderLog.csv");
        DataStreamSource<String> payDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\ReceiptLog.csv");

        SingleOutputStreamOperator<OrderEvent> orderInfoDS = orderDS
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String s) throws Exception {
                        String[] fields = s.split(",");
                        return new OrderEvent(
                                Long.parseLong(fields[0]),
                                fields[1],
                                fields[2],
                                Long.parseLong(fields[3])
                        );
                    }
                })
                .filter(ele->!"".equals(ele.getTxId()));

        SingleOutputStreamOperator<TxEvent> txEventDS = payDS.map(new MapFunction<String, TxEvent>() {
            @Override
            public TxEvent map(String value) throws Exception {
                String[] fields = value.split(",");
                return new TxEvent(
                        fields[0],
                        fields[1],
                        Long.parseLong(fields[2])
                );
            }
        });

        // 数据处理
        ConnectedStreams<OrderEvent, TxEvent> connDS = orderInfoDS.connect(txEventDS)
                .keyBy(order -> order.getOrderId(), tx -> tx.getTxId());

        connDS.process(new CoProcessFunction<OrderEvent, TxEvent, String>() {
            // 用于存放订单数据
            Map<String, OrderEvent> orderMap = new HashMap<>();
            // 用于存放支付信息
            Map<String, TxEvent> txMap = new HashMap<>();

            @Override
            public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
                if (txMap.containsKey(value.getTxId())) {
                    out.collect("订单" + value + "= 对账成功");
                    txMap.remove(value.getTxId());
                } else {
                    orderMap.put(value.getTxId(), value);
                }
            }

            @Override
            public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                if (orderMap.containsKey(value.getTxId())) {
                    out.collect("订单：" + value + "=对账成功");
                    orderMap.remove(value.getTxId());
                } else {
                    txMap.put(value.getTxId(), value);
                }


            }
        }).print();


        // 运行执行
        env.execute();

    }
}
